package com.cmfi.utils;

import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.atomic.AtomicInteger;

import org.eclipse.milo.opcua.sdk.client.OpcUaClient;
import org.eclipse.milo.opcua.sdk.client.api.identity.AnonymousProvider;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscription;
import org.eclipse.milo.opcua.sdk.client.api.subscriptions.UaSubscriptionManager;
import org.eclipse.milo.opcua.sdk.client.nodes.UaNode;
import org.eclipse.milo.opcua.sdk.client.subscriptions.ManagedDataItem;
import org.eclipse.milo.opcua.sdk.client.subscriptions.ManagedSubscription;
import org.eclipse.milo.opcua.stack.core.AttributeId;
import org.eclipse.milo.opcua.stack.core.Identifiers;
import org.eclipse.milo.opcua.stack.core.UaException;
import org.eclipse.milo.opcua.stack.core.security.SecurityPolicy;
import org.eclipse.milo.opcua.stack.core.types.builtin.DataValue;
import org.eclipse.milo.opcua.stack.core.types.builtin.DateTime;
import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode;
import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
import org.eclipse.milo.opcua.stack.core.types.structured.MonitoredItemCreateRequest;
import org.eclipse.milo.opcua.stack.core.types.structured.MonitoringParameters;
import org.eclipse.milo.opcua.stack.core.types.structured.ReadValueId;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import com.cmfi.domain.SubscribeCallBack;
import com.cmfi.domain.PlcEntity.PlcNode;

@Component
public class OpcUaUtil {
	private static ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
	static {
		// 核心线程数：线程池中始终保留的线程数量
		executor.setCorePoolSize(5);
		// 最大线程数：线程池允许的最大线程数
		executor.setMaxPoolSize(10);
		// 队列容量：线程池所使用的任务队列大小
		executor.setQueueCapacity(20);
		// 线程存活时间：非核心线程闲置超过此时间后会被回收
		executor.setKeepAliveSeconds(30);
		// 线程名前缀：用于区分线程池中的线程
		executor.setThreadNamePrefix("PlcCallBack-thread-");
		// 拒绝策略：当线程池无法接受新任务时的应对策略
		executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
		// 初始化线程池
		executor.initialize();
	}

	static Logger log = LoggerFactory.getLogger(OpcUaUtil.class);

	//订阅事件Latch，使得某线程在countDown之前一直监听PLC订阅事件
	public static CountDownLatch eventLatch = new CountDownLatch(1);

	/**
	 * 创建OPC UA客户端
	 */
    public static OpcUaClient createClient(String endPointUrl) throws Exception {
        //opc ua服务端地址
        Path securityTempDir = Paths.get(System.getProperty("java.io.tmpdir"), "security");
        Files.createDirectories(securityTempDir);
        if (!Files.exists(securityTempDir)) {
            throw new Exception("unable to create security dir: " + securityTempDir);
        }
        return OpcUaClient.create(
            endPointUrl,
            endpoints -> endpoints.stream().filter(e -> e.getSecurityPolicyUri().equals(SecurityPolicy.None.getUri())).findFirst(),
            configBuilder ->
                configBuilder
                    .setApplicationName(LocalizedText.english("eclipse milo opc-ua client"))
                    .setApplicationUri("urn:eclipse:milo:examples:client")
                    //访问方式
                    .setIdentityProvider(new AnonymousProvider())
                    .setRequestTimeout(UInteger.valueOf(3000))
                    .build()
        );
    }

    /**
     * 遍历树形节点
     *
     * @param client OPC UA客户端
     * @param uaNode 节点
     * @throws Exception
     */
    public static void browseNode(OpcUaClient client, UaNode uaNode) throws Exception {
        List<? extends UaNode> nodes;
        if (uaNode == null) {
            nodes = client.getAddressSpace().browseNodes(Identifiers.ObjectsFolder);
        } else {
            nodes = client.getAddressSpace().browseNodes(uaNode);
        }
        for (UaNode nd : nodes) {
            //排除系统行性节点，这些系统性节点名称一般都是以"_"开头
            if (Objects.requireNonNull(nd.getBrowseName().getName()).startsWith("_")) {
                continue;
            }
            System.out.println("Node= " + nd.getBrowseName().getName() + " " + nd.getNodeId());

            browseNode(client, nd);
        }
    }

   /**
    * Description 读取String类型的node
    * @param client
    * @param identifier
    * @param namespaceIndex
    * @return
    * @throws Exception
    */
    public static Variant readNode(OpcUaClient client, String identifier,int namespaceIndex) throws Exception {
        //节点
        NodeId nodeId = new NodeId(namespaceIndex, identifier);
        //读取节点数据
        DataValue value = client.readValue(0.0, TimestampsToReturn.Neither, nodeId).get();
        if (value == null || value.getValue() == null || value.getValue().getValue() == null) {
            return null;
        }
        return value.getValue();
    }
    
    /**
	 * Description 写入String类型node数据
	 * 
	 * @param client
	 * @param identifier
	 * @param namespaceIndex
	 * @return
	 * @throws Exception
	 */
	public static boolean writeNode(OpcUaClient client, String identifier, int namespaceIndex, Object newValue)
			throws Exception {
		try {
			// 节点
			NodeId nodeId = new NodeId(namespaceIndex, identifier);
			// 读取节点数据
			DataValue value = new DataValue(new Variant(newValue));
			CompletableFuture<StatusCode> writeFuture = client.writeValue(nodeId, value);
			writeFuture.get();
			return true;
		} catch (Exception e) {
			log.info("[OpcUaUtil][writeStringNode] e,", e);
			return false;
		}
	}

	/**
	 * Description 写入String类型node数据
	 * 
	 * @param client
	 * @param identifier
	 * @param namespaceIndex
	 * @return
	 * @throws Exception
	 */
	public static boolean batchWriteNode(OpcUaClient client, Map<NodeId, DataValue> nodeDataMap) throws Exception {
		try {
			List<NodeId> nodes = new ArrayList<NodeId>();
			List<DataValue> datas = new ArrayList<DataValue>();
			for (Map.Entry<NodeId, DataValue> entry : nodeDataMap.entrySet()) {
				nodes.add(entry.getKey());
				datas.add(entry.getValue());
			}
			CompletableFuture<List<StatusCode>> writeFuture = client.writeValues(nodes, datas);
			writeFuture.get();
			return true;
		} catch (Exception e) {
			log.info("[OpcUaUtil][writeNode] e,", e);
			return false;
		}
	}

    /**
     * Description 读取nummeric值
     * @param client
     * @param index
     * @param namespaceIndex
     * @return
     * @throws Exception
     */
	public static Variant readNode(OpcUaClient client, int index, int namespaceIndex) throws Exception {
		// 节点
		NodeId nodeId = new NodeId(namespaceIndex, index);
		// 读取节点数据
		DataValue value = client.readValue(0.0, TimestampsToReturn.Neither, nodeId).get();
		if (value == null || value.getValue() == null || value.getValue().getValue() == null) {
			return null;
		}

		return value.getValue();
	}

	public static List<DataValue> readNodeMulti(OpcUaClient client, List<NodeId> nodes) throws Exception {
		List<DataValue> values = client.readValues(0.0, TimestampsToReturn.Neither, nodes).get();
		return values;
	}

    /**
     * 订阅(单个)
     *
     * @param client
     * @throws Exception
     */
	private static void subscribe(OpcUaClient client, String identifier, int namespaceIndex) throws Exception {
		AtomicInteger a = new AtomicInteger();
		// 创建发布间隔1000ms的订阅对象
		client.getSubscriptionManager().createSubscription(1000.0).thenAccept(t -> {
			// 节点
			NodeId nodeId = new NodeId(namespaceIndex, identifier);
			ReadValueId readValueId = new ReadValueId(nodeId, AttributeId.Value.uid(), null, null);
			// 创建监控的参数
			MonitoringParameters parameters = new MonitoringParameters(UInteger.valueOf(a.getAndIncrement()), 1000.0,
					null, UInteger.valueOf(10), true);
			// 创建监控项请求
			// 该请求最后用于创建订阅。
			MonitoredItemCreateRequest request = new MonitoredItemCreateRequest(readValueId, MonitoringMode.Reporting,
					parameters);
			List<MonitoredItemCreateRequest> requests = new ArrayList<>();
			requests.add(request);
			// 创建监控项，并且注册变量值改变时候的回调函数。
			t.createMonitoredItems(TimestampsToReturn.Both, requests, (item, id) -> item.setValueConsumer((it, val) -> {
				System.out.println("nodeid :" + it.getReadValueId().getNodeId());
				System.out.println("value :" + val.getValue().getValue());
			}));
		}).get();

        //持续订阅
        Thread.sleep(Long.MAX_VALUE);
    }


    /**
     * 处理订阅业务
     *
     * @param client OPC UA客户端
     */
	private static void batchSubscribe(OpcUaClient client, List<String> identifierList, int namespaceIndex,
			SubscribeCallBack callBack) {
		try {
			// 创建订阅
			ManagedSubscription subscription = ManagedSubscription.create(client);
			List<NodeId> nodeIdList = new ArrayList<>();
			for (String s : identifierList) {
				nodeIdList.add(new NodeId(namespaceIndex, s));
			}
			// 监听
			List<ManagedDataItem> dataItemList = subscription.createDataItems(nodeIdList);
			for (ManagedDataItem managedDataItem : dataItemList) {
				managedDataItem.addDataValueListener(t -> {
					executor.execute(() -> {
						log.info("[OpcUaUtil][batchSubscribe]:{},{}",
								managedDataItem.getNodeId().getIdentifier().toString(),
								t.getValue().getValue().toString());
						if (callBack == null) {
							log.info("[OpcUaUtil][batchSubscribe]callBack is null:{},{}",
									managedDataItem.getNodeId().getIdentifier().toString(),
									t.getValue().getValue().toString());
							return;
						}
						callBack.call(managedDataItem.getNodeId().getIdentifier().toString(),
								t.getValue().getValue().toString());
					});
				});
			}
		} catch (Exception e) {
			log.info("[OpcUaUtil][batchSubscribe] e,", e);
		}
	}
	
	/**
	 * 处理订阅业务
	 *
	 * @param client OPC UA客户端
	 */
	private static void batchSubscribe(OpcUaClient client, Set<PlcNode> nodes, SubscribeCallBack callBack) {
		try {
			// 创建订阅
			ManagedSubscription subscription = ManagedSubscription.create(client);
			List<NodeId> nodeIdList = new ArrayList<>();
			for (PlcNode node : nodes) {
				nodeIdList.add(new NodeId(node.getNamespaceIndex(), node.getIdentifier()));
			}
			// 监听
			List<ManagedDataItem> dataItemList = subscription.createDataItems(nodeIdList);
			for (ManagedDataItem managedDataItem : dataItemList) {
				managedDataItem.addDataValueListener(t -> {
					executor.execute(() -> {
						log.info("[OpcUaUtil][batchSubscribe]:{},{}",
								managedDataItem.getNodeId().getIdentifier().toString(),
								t.getValue().getValue().toString());
						if (callBack == null) {
							log.info("[OpcUaUtil][batchSubscribe]callBack is null:{},{}",
									managedDataItem.getNodeId().getIdentifier().toString(),
									t.getValue().getValue().toString());
							return;
						}
						callBack.call(managedDataItem.getNodeId().getIdentifier().toString(),
								t.getValue().getValue().toString());
					});
				});
			}
		} catch (Exception e) {
			log.info("[OpcUaUtil][batchSubscribe] e,", e);
		}
	}

    /**
     * 自定义订阅监听
     */
    private static class CustomSubscriptionListener implements UaSubscriptionManager.SubscriptionListener {

        private OpcUaClient client;
        
        private Set<PlcNode> nodes;
        
        private SubscribeCallBack callBack;


		public CustomSubscriptionListener(OpcUaClient client, Set<PlcNode> nodes, SubscribeCallBack callBack) {
			super();
			this.client = client;
			this.nodes = nodes;
			this.callBack = callBack;
		}

		/**
		 * 心跳
		 */
		public void onKeepAlive(UaSubscription subscription, DateTime publishTime) {
			log.debug("[OpcUaUtil][onKeepAlive]onKeepAlive,subscription:{}", subscription);
        }

        public void onStatusChanged(UaSubscription subscription, StatusCode status) {
        	log.debug("[OpcUaUtil][onStatusChanged]onStatusChanged,subscription:{}", subscription);
        }

        /**
         * 订阅失败时自动调用
         */
        public void onPublishFailure(UaException exception) {
        	log.info("[OpcUaUtil][onPublishFailure]onPublishFailure,exception:{}", exception);
        }

        public void onNotificationDataLost(UaSubscription subscription) {
        	log.debug("[OpcUaUtil][onNotificationDataLost]onPublishFailure,subscription:{}", subscription);
        }

        /**
		 * 重连时 尝试恢复之前的订阅失败时 会调用此方法
		 *
		 * @param uaSubscription 订阅
		 * @param statusCode     状态
		 */
		public void onSubscriptionTransferFailed(UaSubscription uaSubscription, StatusCode statusCode) {
			log.info(
					"[OpcUaUtil][onSubscriptionTransferFailed]onSubscriptionTransferFailed,uaSubscription:{},statusCode:{}",
					uaSubscription, statusCode);
			batchSubscribe(client, nodes, callBack);
		}
	}
    
    /**
	 * 批量订阅,支持断连后，重新连接时自动下发订阅
	 *
	 * @param client
	 * @throws Exception
	 */
	public static void batchSubscribeWithDisconnectReSub(OpcUaClient client, Set<PlcNode> nodeSet, SubscribeCallBack callBack) throws Exception {
		// 添加订阅监听器，用于处理断线重连后的订阅问题
		log.info("[OpcUaUtil][batchSubscribeWithDisconnectReSub] identifierList:{}", nodeSet);
		client.getSubscriptionManager().addSubscriptionListener(
				new CustomSubscriptionListener(client, nodeSet, callBack));
		// 处理订阅业务
		batchSubscribe(client, nodeSet, callBack);
		// 持续监听
		eventLatch.await();
		log.info(
				"[OpcUaUtil][batchSubscribeWithDisconnectReSub] now thread stop monitor,监听结束");
	}
	
	/**
	 * 断开opcUaClient连接
	 * @param opcUaClient OpcUaClient
	 */
	public static void disconnect(OpcUaClient opcUaClient) {
		if (Objects.nonNull(opcUaClient)) {
			try {
				opcUaClient.disconnect().whenComplete((result, ex) -> {
					if (ex != null) {
						log.error(ex.getMessage());
					} else {
						log.info("[OpcUaUtil][disconnect] {} 断开连接成功!", opcUaClient.getConfig().getApplicationUri());
					}
				}).get();
			} catch (Exception e) {
				log.info("[OpcUaUtil][disconnect] e,", e);
			}
		}
	}
	
	/**
	 * 断开opcUaClient连接
	 * @param opcUaClient OpcUaClient
	 */
	public static void shutdownThreadPool() {
		executor.shutdown();
	}


//	public static void main(String[] args) throws Exception {
//		String endPointUrl = "opc.tcp://LAPTOP-SGV9P5OF:53530/OPCUA/SimulationServer";
//		OpcUaClient client = createClient(endPointUrl);
//		client.connect().get();
//		Variant ww = readNode(client, "waiyuanrongfu.dianliu", 3);
//		System.out.println("caijide shuju e:" + ww);
//		
//		writeNode(client, "waiyuanrongfu.dianliu", 3, Double.valueOf(55));
//		Variant ee = readNode(client, "waiyuanrongfu.dianliu", 3);
//		System.out.println("caijide shuju zuizhong:" + ee);
//		List<NodeId> list = new ArrayList<>();
//		NodeId nodeId = new NodeId(3, "energy");
//		NodeId nodeId1 = new NodeId(3, "neibirongfu.dianya");
//		list.add(nodeId);
//		list.add(nodeId);
//		System.out.println("readNodeMulti1:" + readNodeMulti(client, list));
//		Map<NodeId, DataValue> map= new HashMap<>();
//		DataValue value = new DataValue(new Variant(Double.valueOf(444)));
//		map.put(nodeId, value);
//		DataValue value1 = new DataValue(new Variant("555"));
//		map.put(nodeId1, value1);
//		batchWriteNode(client, map);
//		System.out.println("readNodeMulti2:" + readNodeMulti(client, list));
//		
//		//批量订阅和单个订阅都是OK的,断联后重新订阅，也是OK的
//		List<String> listId = new ArrayList<>();
//		listId.add("WORK_ORDER_RELEASE_FINISH_FLAG");
//		batchSubscribeWithDisconnectReSub(client, listId, 3,null);
//	}
	


    /**
     * Description 解析参数值
     * @param v
     * @return Object
     */
	public static Object parseValue(Variant v) {
		if (v == null || v.getValue() == null) {
			return null;
		}
		String objectType = v.getValue().getClass().getSimpleName();
		if (Objects.equals(objectType, "String")) {
			return v.getValue().toString();
		} else if (Objects.equals(objectType, "Integer") || Objects.equals(objectType, "UShort")
				|| Objects.equals(objectType, "Short")) {
			return Integer.parseInt(v.getValue().toString());
		} else if (Objects.equals(objectType, "Double")) {
			return Double.parseDouble(v.getValue().toString());
		} else if (Objects.equals(objectType, "Float")) {
			return Float.parseFloat(v.getValue().toString());
		} else if (Objects.equals(objectType, "Boolean")) {
			return Boolean.parseBoolean(v.getValue().toString());
		} else {
			return null;
		}
	}
}
